package com.hzmg.akkacloud.worker;

import akka.actor.*;
import akka.cluster.Cluster;
import akka.routing.RoundRobinGroup;
import akka.routing.RoundRobinPool;
import akka.routing.SmallestMailboxPool;
import com.google.common.base.Stopwatch;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.hzmg.akkacloud.common.RemoteConstant;
import com.hzmg.akkacloud.common.exception.AkkaCloudException;
import com.hzmg.akkacloud.common.response.ResultDTO;
import com.hzmg.akkacloud.common.serialize.JsonUtils;
import com.hzmg.akkacloud.common.utils.CommonUtils;
import com.hzmg.akkacloud.common.utils.HttpUtils;
import com.hzmg.akkacloud.common.utils.NetUtils;
import com.hzmg.akkacloud.worker.actors.HealthReporterActor;
import com.hzmg.akkacloud.worker.actors.RouterCreatActor;
import com.hzmg.akkacloud.worker.actors.TroubleshootingActor;
import com.hzmg.akkacloud.worker.annotation.creater.RemoteServiceActorCreator;
import com.hzmg.akkacloud.worker.common.AkkaCloudBannerPrinter;
import com.hzmg.akkacloud.worker.common.AkkaCloudWorkerConfig;
import com.hzmg.akkacloud.worker.common.ThreadLocalStore;
import com.hzmg.akkacloud.worker.common.WorkerRuntime;
import com.hzmg.akkacloud.worker.common.utils.SpringUtils;
import com.hzmg.akkacloud.worker.starter.AkkaStarter;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.util.*;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 客户端启动类
 *
 * @author zbw
 * @since 2022/7/7
 */
@Slf4j
public class AkkaCloudWorker implements ApplicationContextAware, InitializingBean, DisposableBean {

    private final WorkerRuntime workerRuntime = new WorkerRuntime();
    private final AtomicBoolean initialized = new AtomicBoolean();
    private ApplicationContext applicationContext;
    //web port
    @Value("${server.port}")
    String webPort;
    private ScheduledExecutorService timingPool;

    @Override
    public void setApplicationContext(@NotNull ApplicationContext applicationContext) throws BeansException {
        SpringUtils.inject(applicationContext);
        this.applicationContext = applicationContext;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        init();
    }

    public void init() throws Exception {
        //确保只运行一次
        if (!initialized.compareAndSet(false, true)) {
            log.warn("[AkkaCloud-Worker]: please do not repeat the initialization");
            return;
        }
        //计时
        Stopwatch stopwatch = Stopwatch.createStarted();
        log.info("[AkkaCloud-Worker] start to initialize AkkaCloudWorker...");

        AkkaCloudWorkerConfig config = workerRuntime.getWorkerConfig();
        CommonUtils.requireNonNull(config, "can't find WorkerConfig, please set WorkerConfig first");

        try {
            AkkaCloudBannerPrinter.print();
            // 注册服务到注册中心
            if (!config.isEnableTestMode()) {
                register();
            } else {
                log.info("[AkkaCloudWorkerProperties]:" + config);
                log.warn("[AkkaCloudWorker] using TestMode now, it's dangerous if this is production env.");
            }
            //获取cpu的个数
            int cores = Runtime.getRuntime().availableProcessors();
            // 初始化元数据
            //akka Ip
            String workerAddress = NetUtils.getLocalHost() + ":" + config.getPort();
            workerRuntime.setWorkerAddress(workerAddress);
            //web Ip
            String webAddress = NetUtils.getLocalHost() + ":" + webPort;
            workerRuntime.setWebAddress(webAddress);

            // 初始化定时线程池
            ThreadFactory timingPoolFactory = new ThreadFactoryBuilder().setNameFormat("oms-worker-timing-pool-%d").build();
            //timingPool = Executors.newScheduledThreadPool(cores, timingPoolFactory);

            timingPool = new ScheduledThreadPoolExecutor(cores, timingPoolFactory);
            ActorSystem actorSystem=AkkaStarter.init(config);
            //初始化汇报心跳actor
            ActorRef healthyReportRef = actorSystem.actorOf(HealthReporterActor.props(workerRuntime).withRouter(new RoundRobinPool(2)));
            // 上报健康信息 同时更新可用服务列表
            timingPool.scheduleAtFixedRate(() -> healthyReportRef.tell("go", ActorRef.noSender()), 1, 25, TimeUnit.SECONDS);

            //创建远程服务调用
            RemoteServiceActorCreator.initRemoteServiceInvoke(applicationContext, actorSystem);


            //创建路由actor
            actorSystem.actorOf(Props.create(RouterCreatActor.class).withRouter(new SmallestMailboxPool(cores*2)),RemoteConstant.ROUTER_CREATER_ACTOR_NAME);
            // 发送失败的消息处理
           /* ActorRef troubleshootingActor = actorSystem.actorOf(Props.create(TroubleshootingActor.class), RemoteConstant.TROUBLESHOOTING_ACTOR_NAME);
            actorSystem.eventStream().subscribe(troubleshootingActor, DeadLetter.class);*/
            workerRuntime.setActorSystem(actorSystem);
            log.info("[AkkaCloudWorker] akka-remote listening address: {}", workerAddress);
            log.info("[AkkaCloudWorker] akka ActorSystem({}) initialized successfully.", actorSystem);

 /*           // 初始化日志系统 里面的作用以后将会更换
            OmsLogHandler omsLogHandler = new OmsLogHandler(workerAddress, actorSystem, serverDiscoveryService);
            workerRuntime.setOmsLogHandler(omsLogHandler);
*/

            log.info("[AkkaCloudWorker] local storage initialized successfully.");

               /*//定时上报日志
            timingPool.scheduleWithFixedDelay(omsLogHandler.logSubmitter, 0, 5, TimeUnit.SECONDS);
            timingPool.scheduleWithFixedDelay(omsLogHandler.logSubmitter, 0, 5, TimeUnit.SECONDS);*/

            //存进线程内存中
            ThreadLocalStore.setRuntimeMeta(workerRuntime);
            log.info("[AkkaCloudWorker] AkkaCloudWorker initialized successfully, using time: {}, congratulations!", stopwatch);
        } catch (Exception e) {
            log.error("[AkkaCloudWorker] initialize AkkaCloudWorker failed, using {}.", stopwatch, e);
            throw e;
        }
    }

    public void setConfig(AkkaCloudWorkerConfig config) {
        workerRuntime.setWorkerConfig(config);
    }

    /**
     * Register the service to the registry.
     */
    @SuppressWarnings("rawtypes")
    private void register() {
        AkkaCloudWorkerConfig config = workerRuntime.getWorkerConfig();
        String appName = config.getAppName();
        Objects.requireNonNull(appName, "appName can't be empty!");

        String url = "http://%s/server/register?appName=%s";
        for (String server : config.getServerAddress()) {
            String realUrl = String.format(url, server, appName);
            try {
                String resultDTOStr = CommonUtils.executeWithRetry0(() -> HttpUtils.get(realUrl));
                ResultDTO resultDTO = JsonUtils.parseObject(resultDTOStr, ResultDTO.class);
                if (resultDTO.isSuccess()) {
                    Long appId = Long.valueOf(resultDTO.getData().toString());
                    log.info("[AkkaCloudWorker] assert appName({}) succeed, the appId for this application is {}.", appName, appId);
                    workerRuntime.setAppId(appId);
                    //workerRuntime.setFirstConsoleAddress(server + ":" + getPort(server));
                    return;
                } else {
                    throw new AkkaCloudException(resultDTO.getMessage());
                }
            } catch (AkkaCloudException oe) {
                throw oe;
            } catch (Exception ignore) {
                log.warn("[AkkaCloudWorker] register appName by url({}) failed, please check the server address.", realUrl);
            }

        }
        log.error("[AkkaCloudWorker] no available server in {}.", config.getServerAddress());
        throw new AkkaCloudException("no server available!");
    }

    /**
     * Obtain the Port(actorSystem) of the registry。
     */
    private int getPort(String serverIP) {
        AkkaCloudWorkerConfig config = workerRuntime.getWorkerConfig();
        return visitWeb(serverIP, config, log);
    }

    public static int visitWeb(String serverIP, AkkaCloudWorkerConfig config, Logger log) {
        String appName = config.getAppName();
        String url = "http://%s/server/getPort?appName=%s";
        String realUrl = String.format(url, serverIP, appName);
        try {
            String resultDTOStr = CommonUtils.executeWithRetry0(() -> HttpUtils.get(realUrl));
            ResultDTO resultDTO = JsonUtils.parseObject(resultDTOStr, ResultDTO.class);
            if (resultDTO.isSuccess()) {
                int port = Integer.parseInt(resultDTO.getData().toString());
                log.info("[AkkaCloudWorker] get the port of actorSystem succeed, the port for this register is {}.", port);
                return port;
            } else {
                throw new AkkaCloudException(resultDTO.getMessage());
            }
        } catch (AkkaCloudException oe) {
            throw oe;
        }catch (Exception e){
            e.printStackTrace();
            log.warn("[AkkaCloudWorker] find the registry port by url({}) failed, please check the server address.", realUrl);
            log.error("[AkkaCloudWorker] can not find this registry akka port {}.", serverIP);
            throw new AkkaCloudException("[AkkaCloudWorker] Maybe a registry has not been started,please check the configuration");
        }

    }

    @Override
    public void destroy() throws Exception {
        timingPool.shutdownNow();
        //关闭ActorSystem
        workerRuntime.getActorSystem().terminate();
    }
}
